
#include <unistd.h>
#include <cfloat>
#include "base/logging.h"
#include "base/scoped_ptr.h"
#include "net/rpc/rpc.h"
#include "net/rpc/rpc_channel.h"
#include "thread/executor.h"
#include <google/protobuf/message.h>
#include <google/protobuf/descriptor.h>
#include "net/rpc/rpcmessage.pb.h"

using google::protobuf::Message;
using google::protobuf::MethodDescriptor;
using google::protobuf::RpcController;

RpcClientConnection::RpcClientConnection(const char* ip, int port, Executor* executor)
    : AsyncClientConnection(ip, port),
      executor_(executor) {
}
  
void RpcClientConnection::HandleRecv(char* buffer, int size) {
  executor_->Add(NewCallback(this, &RpcClientConnection::HandleRecvInternal, buffer, size));
}

void RpcClientConnection::HandleRecvInternal(char* buffer, int size) {
  scoped_ptr<RpcMessageHeader> respheader;
  int message_size;
  scoped_array<char> tmp;
  {
    WriteLock wl(&recv_mutex_);
    read_buf.Push(buffer, size);
    int header_size;
    if (!read_buf.PeekTwoInt(&header_size, &message_size)) return;
    if (read_buf.size() < sizeof(int) * 2 + header_size + message_size) return;
    char nosense[sizeof(int) * 2];
    read_buf.Pop(nosense, sizeof(int) * 2);
    tmp.Reset(new char[std::max(message_size, header_size)]);
    read_buf.Pop(tmp.get(), header_size);
    respheader.Reset(new RpcMessageHeader);
    respheader->ParseFromArray(tmp.get(), header_size);
    read_buf.Pop(tmp.get(), message_size);
  }
  FinalizeRPC(respheader->id(), tmp.get(), message_size);
}
  
void RpcClientConnection::FinalizeRPC(int64 req_id,
    char* buffer, int buffer_size) {
  Closure* callback = NULL;
  {
    WriteLock wl(&rpc_context_mutex_);
    if (rpc_context_.find(req_id) != rpc_context_.end()) {
      callback = rpc_context_[req_id].closure;
      if (buffer == NULL) {
        VLOG(1) << "deadline exceeded";
        rpc_context_[req_id].rpc->SetStatus(RPC::DEADLINE_EXCEEDED);
      } else {
        rpc_context_[req_id].msg->ParseFromArray(buffer, buffer_size);
      }
      rpc_context_.erase(req_id);
    }
  }
  if (callback != NULL) {
    callback->Run();
  }
}
void RpcClientConnection::HandleDisconnect() {
  status_ = DISCONNECTED;
}

void RpcClientConnection::DeadlineExceeded(int64 req_id) {
  FinalizeRPC(req_id, NULL, 0);
}

// TODO(hjbai): Should return immediately.
void RpcClientConnection::SendMessage(
    RPC* rpc,
    int64* req_id, const string& name,
    const Message& req, Message* resp, Closure* callback) {
  string* s1 = new string;
  string* s2 = new string;
  {
   WriteLock wl(&rpc_context_mutex_);
   RpcMessageHeader m;
   m.set_name(name);
   m.set_id(*req_id);
   if (rpc->deadline() != FLT_MAX) {
     executor_->AddAfter(rpc->deadline() * 1000000, NewCallback(this, &RpcClientConnection::DeadlineExceeded, *req_id));
   }
   RpcContext context;
   context.closure = callback;
   context.msg = resp;
   context.rpc = rpc;
   rpc_context_[*req_id] = context;
   ++(*req_id);
   *s1 = m.SerializeAsString();
   *s2 = req.SerializeAsString();
  }
  SendInternal(s1, s2);
}

void RpcClientConnection::SendInternal(string* s1, string* s2) {
  int size = s1->size();
  char buf[sizeof(int)];
  memcpy(buf, (char*)(&size), sizeof(int));
  Send(buf, sizeof(int));
  size = s2->size();
  memcpy(buf, (char*)(&size), sizeof(int));
  Send(buf, sizeof(int));
  Send(s1->c_str(), s1->size());
  Send(s2->c_str(), s2->size());
  delete s1;
  delete s2;
}

ClientRpcChannel::ClientRpcChannel(const char* ip, int port, ClientPollRunner* runner,
    Executor* executor)
    : conn_(ip, port, executor) {
  conn_.SetPollRunner(runner);
  i_ = 100;
  conn_.Connect();
  LoopThread* t = new LoopThread(runner);
  t->Start();
}

void ClientRpcChannel::CallMethod(const MethodDescriptor* method,
    RpcController* controller,
    const Message* request,
    Message* response,
    Closure* done) {
  RPC* rpc = (RPC*) controller;
  CHECK_EQ(rpc->status(), RPC::PENDING);
  conn_.SendMessage(rpc, &i_, method->name(), *request, response, done);
}

